package xyj.sparrow.scheduler.cron.thread;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import xyj.sparrow.scheduler.cron.Cron;
import xyj.sparrow.scheduler.cron.CronComparator;
import xyj.sparrow.scheduler.entity.Task;
import xyj.sparrow.scheduler.mapper.TaskMapper;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 监听线程
 * 监听所有任务
 *
 * @author 江南小俊
 * @since 2021/03/03 20:58:00
 */
@Slf4j
@Component()
public class MonitorThread implements Runnable {
    private final ReentrantLock reentrantLock = new ReentrantLock();
    private final Condition condition = reentrantLock.newCondition();
    private final ConcurrentSkipListSet<Cron> cronSet = new ConcurrentSkipListSet<>(new CronComparator());
    private final ConcurrentHashMap<String, Boolean> updateFlag = new ConcurrentHashMap<>();
    private final ReentrantLock updateLock = new ReentrantLock();

    @Resource
    private SchedulingThreadPool schedulingThreadPool;
    @Resource
    private TaskMapper taskMapper;

    /**
     * 启动监控线程
     */
    @PostConstruct
    private void init() {
        log.info("MonitorThread is init...");
        //加载任务集合
        loadTaskCollection();
        start();
    }

    public void start() {
        Thread thread = new Thread(this);
        thread.setDaemon(true);
        thread.setName("monitorThread");
        thread.start();
    }

    /**
     * 加载任务集合
     */
    private void loadTaskCollection() {
        //从数据库库读取
        List<Task> tasks = taskMapper.listTaskOfEnable();
        tasks.forEach(task -> {
            Cron cron = new Cron(this, task);
            cronSet.add(cron);
        });

    }

    public void addCron(Cron cron) {
        updateLock.lock();
        log.info("addCron:{}", cron);
        if (updateFlag.containsKey(cron.getTask().getTaskId()) && updateFlag.get(cron.getTask().getTaskId())) {
            //已更新，这个旧的cron不需要更新
            //这里只会重新排序set,不会替换cron
            cronSet.add(cron);
            updateFlag.put(cron.getTask().getTaskId(), false);
        } else {
            cronSet.removeIf(cronB ->
                    cronB.getTask().getTaskId().equals(cron.getTask().getTaskId())
            );
            cronSet.add(cron);
        }

        signalAll();
        updateLock.unlock();
    }

    public void updateCron(Cron cron) {
        updateLock.lock();
        updateFlag.put(cron.getTask().getTaskId(), true);
        log.info("updateCron:{}", cron);
        cronSet.removeIf(cronB ->
                cronB.getTask().getTaskId().equals(cron.getTask().getTaskId())
        );
        cronSet.add(cron);
        signalAll();
        updateLock.unlock();
    }

    @Override
    public void run() {
        log.info("MonitorThread is running...");
        while (true) {
            if (cronSet.isEmpty()) {
                try {
                    reentrantLock.lock();
                    condition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    reentrantLock.unlock();
                }
                continue;
            }
            Cron first = cronSet.first();

            long interval = first.getCronExpression().getNextExecutionTime().getTime() - new Date().getTime();
            if (Math.abs(interval) < 300) {
                first = cronSet.pollFirst();
                if (first == null) continue;
                first.setNow(LocalDateTime.now());
                schedulingThreadPool.execute(first);
            } else {
                if (interval < 0) {
                    Cron cron = cronSet.pollFirst();
                    if (cron == null) continue;
                    //过期的
                    addCron(cron.refreshTime());
                } else {
                    try {
                        reentrantLock.lock();
                        condition.await(interval > 0 ? interval : 0, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        reentrantLock.unlock();
                    }
                }

            }

        }
    }

    /**
     * 唤醒线程
     *
     * @author 努力努力再努力v
     * @since 2021/6/14
     */


    public void signalAll() {
        reentrantLock.lock();
        condition.signalAll();
        reentrantLock.unlock();
    }

    public void removeByTaskId(String taskId) {
        log.info("removeByTaskId:{}", taskId);
        cronSet.removeIf(next -> next.getTask().getTaskId().equals(taskId));
        signalAll();
    }

    public void removeByTaskIds(List<String> taskIds) {
        log.info("removeByTaskIds:{}", taskIds);
        cronSet.removeIf(next -> taskIds.contains(next.getTask().getTaskId()));
        signalAll();
    }

    public void removeByAppId(String appId) {
        log.info("removeById:{}", appId);
        cronSet.removeIf(next -> next.getTask().getAppId().equals(appId));
        signalAll();
    }

    public void removeByAppIds(List<String> appIds) {
        log.info("removeByIds:{}", appIds);
        cronSet.removeIf(next -> appIds.contains(next.getTask().getAppId()));
        signalAll();
    }
}
